package cn.sheep.dmp.etl

import cn.sheep.dmp.utils.Tools
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 原始日志转换成parquet文件
  * Sheep.Old @ 64341393
  * Created 2018/3/28
  */
object Biz2Parquet {

  def main(args: Array[String]): Unit = {

    // 接受参数
    if (args.length != 3) {

      println(
        """
          |cn.sheep.dmp.etl.Biz2Parquet
          |参数：
          |     dataInputPath <原始日志所在路径>
          |     compressionCode <压缩格式>
          |     outputPath  <转换后的日志存储路径>
        """.stripMargin)
      sys.exit()
    }

    val Array(dataInputPath, compressionCode, outputPath) = args

    val sparkConf = new SparkConf().setAppName("日志转parquet文件")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(sparkConf)

    // 读取数据
    val filtered: RDD[Array[String]] = sc.textFile(dataInputPath).map(_.split(",", -1)).filter(_.length >= 85)
    val sqlc = new SQLContext(sc)

    // 日志转parquet <= schema + RDD[Row]
    val row = filtered.map(arr => {
      Row(
        arr(0),
        Tools.str2Int(arr(1)),
        Tools.str2Int(arr(2)),
        Tools.str2Int(arr(3)),
        Tools.str2Int(arr(4)),
        arr(5),
        arr(6),
        Tools.str2Int(arr(7)),
        Tools.str2Int(arr(8)),
        Tools.str2Double(arr(9)),
        Tools.str2Double(arr(10)),
        arr(11),
        arr(12),
        arr(13),
        arr(14),
        arr(15),
        arr(16),
        Tools.str2Int(arr(17)),
        arr(18),
        arr(19),
        Tools.str2Int(arr(20)),
        Tools.str2Int(arr(21)),
        arr(22),
        arr(23),
        arr(24),
        arr(25),
        Tools.str2Int(arr(26)),
        arr(27),
        Tools.str2Int(arr(28)),
        arr(29),
        Tools.str2Int(arr(30)),
        Tools.str2Int(arr(31)),
        Tools.str2Int(arr(32)),
        arr(33),
        Tools.str2Int(arr(34)),
        Tools.str2Int(arr(35)),
        Tools.str2Int(arr(36)),
        arr(37),
        Tools.str2Int(arr(38)),
        Tools.str2Int(arr(39)),
        Tools.str2Double(arr(40)),
        Tools.str2Double(arr(41)),
        Tools.str2Int(arr(42)),
        arr(43),
        Tools.str2Double(arr(44)),
        Tools.str2Double(arr(45)),
        arr(46),
        arr(47),
        arr(48),
        arr(49),
        arr(50),
        arr(51),
        arr(52),
        arr(53),
        arr(54),
        arr(55),
        arr(56),
        Tools.str2Int(arr(57)),
        Tools.str2Double(arr(58)),
        Tools.str2Int(arr(59)),
        Tools.str2Int(arr(60)),
        arr(61),
        arr(62),
        arr(63),
        arr(64),
        arr(65),
        arr(66),
        arr(67),
        arr(68),
        arr(69),
        arr(70),
        arr(71),
        arr(72),
        Tools.str2Int(arr(73)),
        Tools.str2Double(arr(74)),
        Tools.str2Double(arr(75)),
        Tools.str2Double(arr(76)),
        Tools.str2Double(arr(77)),
        Tools.str2Double(arr(78)),
        arr(79),
        arr(80),
        arr(81),
        arr(82),
        arr(83),
        Tools.str2Int(arr(84))
      )
    })


    // 构建schema
    val schema = StructType(
      Seq(
        StructField("sessionid", StringType),
        StructField("advertisersid", IntegerType),
        StructField("adorderid", IntegerType),
        StructField("adcreativeid", IntegerType),
        StructField("adplatformproviderid", IntegerType),
        StructField("sdkversion", StringType),
        StructField("adplatformkey", StringType),
        StructField("putinmodeltype", IntegerType),
        StructField("requestmode", IntegerType),
        StructField("adprice", DoubleType),
        StructField("adppprice", DoubleType),
        StructField("requestdate", StringType),
        StructField("ip", StringType),
        StructField("appid", StringType),
        StructField("appname", StringType),
        StructField("uuid", StringType),
        StructField("device", StringType),
        StructField("client", IntegerType),
        StructField("osversion", StringType),
        StructField("density", StringType),
        StructField("pw", IntegerType),
        StructField("ph", IntegerType),
        StructField("long", StringType),
        StructField("lat", StringType),
        StructField("provincename", StringType),
        StructField("cityname", StringType),
        StructField("ispid", IntegerType),
        StructField("ispname", StringType),
        StructField("networkmannerid", IntegerType),
        StructField("networkmannername", StringType),
        StructField("iseffective", IntegerType),
        StructField("isbilling", IntegerType),
        StructField("adspacetype", IntegerType),
        StructField("adspacetypename", StringType),
        StructField("devicetype", IntegerType),
        StructField("processnode", IntegerType),
        StructField("apptype", IntegerType),
        StructField("district", StringType),
        StructField("paymode", IntegerType),
        StructField("isbid", IntegerType),
        StructField("bidprice", DoubleType),
        StructField("winprice", DoubleType),
        StructField("iswin", IntegerType),
        StructField("cur", StringType),
        StructField("rate", DoubleType),
        StructField("cnywinprice", DoubleType),
        StructField("imei", StringType),
        StructField("mac", StringType),
        StructField("idfa", StringType),
        StructField("openudid", StringType),
        StructField("androidid", StringType),
        StructField("rtbprovince", StringType),
        StructField("rtbcity", StringType),
        StructField("rtbdistrict", StringType),
        StructField("rtbstreet", StringType),
        StructField("storeurl", StringType),
        StructField("realip", StringType),
        StructField("isqualityapp", IntegerType),
        StructField("bidfloor", DoubleType),
        StructField("aw", IntegerType),
        StructField("ah", IntegerType),
        StructField("imeimd5", StringType),
        StructField("macmd5", StringType),
        StructField("idfamd5", StringType),
        StructField("openudidmd5", StringType),
        StructField("androididmd5", StringType),
        StructField("imeisha1", StringType),
        StructField("macsha1", StringType),
        StructField("idfasha1", StringType),
        StructField("openudidsha1", StringType),
        StructField("androididsha1", StringType),
        StructField("uuidunknow", StringType),
        StructField("userid", StringType),
        StructField("iptype", IntegerType),
        StructField("initbidprice", DoubleType),
        StructField("adpayment", DoubleType),
        StructField("agentrate", DoubleType),
        StructField("lomarkrate", DoubleType),
        StructField("adxrate", DoubleType),
        StructField("title", StringType),
        StructField("keywords", StringType),
        StructField("tagid", StringType),
        StructField("callbackdate", StringType),
        StructField("channelid", StringType),
        StructField("mediatype", IntegerType)
      )
    )

    val value: RDD[String] = filtered.flatMap(x => x)
    sqlc.setConf("spark.sql.parquet.compression.codec", compressionCode)
    val dataFrame = sqlc.createDataFrame(row, schema)

    // 将dataFrame转存parquet
    dataFrame.write.parquet(outputPath)

    sc.stop()

  }

}
